
package com.lee.job; 
 
import com.alibaba.fastjson.JSON; 
import com.alibaba.fastjson.JSONArray; 
import com.google.common.collect.Lists; 
import com.lee.database.HBaseConnPool; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.source.HbaseSource; 
import com.lee.utils.*; 
import org.apache.commons.collections.CollectionUtils; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.util.Collector; 
import org.apache.hadoop.hbase.Cell; 
import org.apache.hadoop.hbase.CellUtil; 
import org.apache.hadoop.hbase.TableName; 
import org.apache.hadoop.hbase.client.Get; 
import org.apache.hadoop.hbase.client.Put; 
import org.apache.hadoop.hbase.client.Result; 
import org.apache.hadoop.hbase.client.Table; 
import org.apache.hadoop.hbase.util.Bytes; 
 
import java.io.IOException; 
import java.time.Duration; 
import java.time.LocalDateTime; 
import java.util.*; 
import java.util.stream.Collectors; 
 
public class DCountHandleFunction implements FlatMapFunction<List<String>, CkFlowInfo> { 
 
 
    @Override
    public void flatMap(List<String> bdsCkFlowList, Collector<CkFlowInfo> collector)  {
 
        try {
 
            DMapUtils.addIndex();
            DCountUtils.addIndex();
            Logger.debug(DMapUtils.getIndex(),"/",DCountUtils.getIndex(),"DCountHandleFunction -> index + 1, bdsCkFlowList ", bdsCkFlowList);
 
 
 
            //判断是否到了块结尾
            long blockSize = ConstantsUtils.DATA_BLOCK_SIZE;
            if (DMapUtils.getIndex() > blockSize) {
                //等待DMap表不再更新
                long waitTimes = 60;
                long finishIndex = 0;
                long lastFinishIndex = -1;
                while(waitTimes > 0 && finishIndex != lastFinishIndex){
                    waitTimes -= 1;
                    lastFinishIndex = finishIndex;
                    finishIndex = 0;
                    for(int i=0;i<FinishQualifierEnum.values().length;i++){
                        finishIndex += DMapUtils.getFinishIndex(FinishQualifierEnum.values()[i].getCode());
                    }
 
                    Logger.info("DCountHandleFunction ->  表处理未完成,进度 "+finishIndex+"，剩余",waitTimes,"次尝试，等待一段时间后重新获取 ");
                    Thread.sleep(10000);
                }
                //获取DMap所有数据
                List<String> allKey = DMapUtils.getAllKey();
                List<List<String>> needKeys = new ArrayList<>();
                List<List<String>> tmpKeys = new ArrayList<>();
                double minBlockSupport = ConstantsUtils.MIN_BLOCK_SUPPORT;
                Logger.info("DCountHandleFunction -> 查询得到dmap表一共有key的数量", allKey.size());
                //先过滤一次
                List<List<String>> allKeys = new ArrayList<>();
                for (String key : allKey) {
                    List<String> countKey = JSONArray.parseArray(key).toJavaList(String.class);

                    //获取key的值，计算支持度计数
                    List<List<Long>> valueByKey = DMapUtils.getValueByKeyGinix(key);
                    long count =  ArrayUtils.getItemSizeGinix(valueByKey);
                    double support = 1.0 * count / blockSize;
                    Logger.debug("DCountHandleFunction -> 校验key是否满足支持度，key:", key, "支持度：", support, "最小块支持度：", minBlockSupport);
                    if (support >= minBlockSupport) {
 
                        long preCount = DCountUtils.getValueByKey(countKey);
                        DCountUtils.setValueByKey(countKey, count + preCount);
                        allKeys.add(countKey);
                    }
 
 
                }
                ;
                Logger.info("DCountHandleFunction -> 过滤后dmap表一共有key的数量", allKeys.size());
 
                needKeys.addAll(allKeys);
                for (int i = 2; i <= ConstantsUtils.LEVEL; i++) {
                    Logger.info("DCountHandleFunction -> 当前级别", i, "allkeys", allKeys.size(), "needKeys", needKeys.size());
                    Map<String, Integer> calculatedMap = new HashMap<>();
                    for (List<String> key : allKeys) {
 
                        for (List<String> need : needKeys) {
                            List<List<Long>> sameIndex = DMapUtils.getValueByKeyGinix(key);
                            List<String> uniqItem = ArrayUtils.getUniqItem(key, need);
                            uniqItem.sort(String::compareTo);
                            if (!calculatedMap.containsKey(uniqItem.toString()) && uniqItem.size() == i) {
 
                                for (int j = 0; j < need.size(); j++) {
                                    List<List<Long>> nextItem = DMapUtils.getValueByKeyGinix(Lists.newArrayList(need.get(j)));
                                    sameIndex = ArrayUtils.getSameItemGinix(sameIndex, nextItem);
                                }
                                long count = ArrayUtils.getItemSizeGinix(sameIndex);
 
                                double support = 1.0 * count / blockSize;
                                Logger.debug("DCountHandleFunction -> 校验key是否满足支持度，key:", uniqItem, "支持度：", support, "最小块支持度：", minBlockSupport);
 
                                if (support >= minBlockSupport) {
                                    List<String> countKey = uniqItem;
                                    long preCount = DCountUtils.getValueByKey(countKey);
                                    DCountUtils.setValueByKey(countKey, count + preCount);
                                    tmpKeys.add(uniqItem);
                                }
                                calculatedMap.put(uniqItem.toString(), 1);
 
 
                            }
                        }
 
                    }
                    needKeys.clear();
                    needKeys.addAll(tmpKeys);
                    tmpKeys.clear();
                }
 
                long dCountL = DCountUtils.getIndex() / 10000;
                double dCountD = 0.5 - 1.0 * dCountL / 10;
                Logger.info(DCountUtils.getIndex(),dCountD,"DCountHandleFunction -> 修改前支持度", ConstantsUtils.MIN_SUPPORT);
                ConstantsUtils.MIN_SUPPORT = dCountD;
                Logger.info(DCountUtils.getIndex(),"DCountHandleFunction -> 当前支持度", ConstantsUtils.MIN_SUPPORT);
 
 
 
                DMapUtils.initMap();
                SupportDataUtils.initMap();
                DMapUtils.addIndex();
 
            }
            CkFlowInfo ckFlowInfo = new CkFlowInfo();
            long index = DMapUtils.getIndex();
            long cindex = DCountUtils.getIndex();
            ckFlowInfo.setDIndex(index);
            ckFlowInfo.setDCIndex(cindex);
            ckFlowInfo.setCkList(bdsCkFlowList);
            collector.collect(ckFlowInfo);
            Logger.info(DMapUtils.getIndex(),"/",DCountUtils.getIndex(),"END DCountHandleFunction -> index + 1, bdsCkFlowList ", bdsCkFlowList);
 
        }catch (Exception e){
            Logger.error("DCountHandleFunction -> 初始化数据报错：",e);
            e.printStackTrace();
        }
    }
 
 
 
} 
